package com.hdd.starter.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.hdd.starter.properties.CanalProperties;
import com.hdd.starter.transfer.DefaultMessageTransponder;
import com.hdd.starter.transfer.TransponderFactory;
import org.apache.commons.lang.StringUtils;

import java.net.InetSocketAddress;
import java.util.Objects;

/**
 * Canal 客户端抽象类
 */
public abstract class AbstractCanalClient {
	
	/**
	 * 运行状态
	 */
	private volatile boolean running;
	
	/**
	 * canal 配置
	 */
	private CanalProperties canalProperties;
	
	
	/**
	 * 构造方法，初始化 canal 的配置以及转换信息的工厂实例
	 *
	 * @param canalPropertie
	 * @return
	 */
	protected AbstractCanalClient(CanalProperties canalPropertie) {
		//参数校验
		Objects.requireNonNull(canalPropertie, "canalConfig 不能为空!");
		//初始化配置
		this.canalProperties = canalPropertie;
		this.factory = (connector, config,listeners, annoListeners) -> new DefaultMessageTransponder(connector, config, listeners, annoListeners);
	}

	/**
	 * 转换工厂类
	 */
	protected final TransponderFactory factory;

	/**
	 * 开启 canal 客户端
	 */
	public void start() {
		CanalProperties canalProperties = getConfig();
		process(processInstanceEntry(canalProperties), canalProperties);
	}
	
	/**
	 * 初始化 canal 连接
	 */
	protected abstract void  process(CanalConnector connector, CanalProperties  canalProperties);
	
	/**
	 * 处理 canal 连接实例
	 *
	 */
	private CanalConnector processInstanceEntry(CanalProperties  canalProperties) {

		//声明连接
		CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalProperties.getHost(), canalProperties.getPort()), canalProperties.getKey(), canalProperties.getUserName(), canalProperties.getPassword());
		//canal 连接
		connector.connect();
		if (!StringUtils.isEmpty(canalProperties.getFilter())) {
			//canal 连接订阅，包含过滤规则
			connector.subscribe(canalProperties.getFilter());
		} else {
			//canal 连接订阅，无过滤规则
			connector.subscribe();
		}
		
		//canal 连接反转
		connector.rollback();
		//返回 canal 连接
		return connector;
	}
	
	
	/**
	 * 获取 canal 配置
	 *
	 * @param
	 * @return
	 */
	protected CanalProperties getConfig() {
		//canal 配置
		CanalProperties config = canalProperties;
		return config;
	}
	
	/**
	 * 停止 canal 客户端
	 *
	 * @param
	 * @return
	 */
	public void stop() {
		setRunning(false);
	}
	
	/**
	 * 返回 canal 客户端的状态
	 *
	 * @param
	 * @return
	 */
	public boolean isRunning() {
		return running;
	}
	
	/**
	 * 设置 canal 客户端状态
	 *
	 * @param running
	 * @return
	 */
	private void setRunning(boolean running) {
		this.running = running;
	}
}